package com.bs.flink.sink;

import lombok.extern.log4j.Log4j2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.boot.Banner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;

/**
 * @author bingshao
 * @date 2022/11/10
 **/
@EnableAutoConfiguration
@Log4j2
public class KafkaSink extends RichSinkFunction<String> {

    private ApplicationContext applicationContext;

    private KafkaTemplate<String, Object> kafkaTemplate;

    @Override
    public void open(Configuration parameters) throws Exception {
        log.info("kafkaOpen=========");
        SpringApplication application = new SpringApplication(KafkaSink.class);
        application.setBannerMode(Banner.Mode.OFF);
        this.applicationContext = application.run(new String[]{});
        this.kafkaTemplate = setlListener((KafkaTemplate) applicationContext.getBean("kafkaTemplate"));
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        log.info("kafkaValue==============" + value);
        kafkaTemplate.send("flink-send", value);
    }

    public KafkaTemplate<String, Object> setlListener(KafkaTemplate<String, Object> kafkaTemplate) {

        kafkaTemplate.setProducerListener(new ProducerListener<String, Object>() {

            @Override
            public void onSuccess(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata) {
                String topic = recordMetadata.topic();
                int partition = recordMetadata.partition();
                long offset = recordMetadata.offset();
                log.info("成功====》主题：" + topic + "=====>分区:" + partition + "=====>位移:" + offset);
            }

            @Override
            public void onError(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata, Exception exception) {
                log.error("错误原因==============》" + exception.getMessage());
            }
        });

        return kafkaTemplate;

    }

}
